package com.shujia.stream

import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}

object RoadFlowDay {
  def main(args: Array[String]): Unit = {
    /**
     * 创建flink sql环境
     *
     */
    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .inStreamingMode() //流处理模式
      //.inBatchMode()//批处理模式
      .build()

    val table: TableEnvironment = TableEnvironment.create(settings)

    /**
     * 创建 source 表
     *
     */
    table.executeSql(
      """CREATE TABLE cars (
        |    car STRING,
        |    city_code STRING,
        |    county_code STRING,
        |    card BIGINT,
        |    camera_id STRING,
        |    orientation STRING,
        |    road_id BIGINT,
        |    `time` BIGINT,
        |    speed DOUBLE
        |) WITH (
        |  'connector' = 'kafka',
        |  'topic' = 'cars',
        |  'properties.bootstrap.servers' = 'master:9092',
        |  'properties.group.id' = 'testGroup',
        |  'scan.startup.mode' = 'earliest-offset',
        |  'format' = 'json'
        |)
        |
        |""".stripMargin)

    /**
     * 创建sink 表
     *
     */

    table.executeSql(
      """
        |
        |CREATE TABLE road_flow_day (
        |  `road_id` BIGINT,
        |  `day_id` STRING ,
        |  `flow` BIGINT,
        |  PRIMARY KEY (road_id,`day_id`) NOT ENFORCED
        |) WITH (
        |    'connector' = 'jdbc',
        |    'url' = 'jdbc:mysql://master:3306/car',
        |    'table-name' = 'road_flow_day',
        |    'username' ='root',
        |    'password'='123456'
        |)
        |""".stripMargin)

    /**
     * 实时统计道路车流量，将结果保存到mysql中
     *
     */

    table.executeSql(
      """
        |insert into road_flow_day
        |select
        |road_id,
        |FROM_UNIXTIME(`time`,'YYYY-MM-dd') day_id,
        |count(1) as flow
        |from cars
        |group by
        |road_id,FROM_UNIXTIME(`time`,'YYYY-MM-dd')
        |
        |""".stripMargin)


  }

}
